#include "subprocess.h"

#include <sys/prctl.h>
#include <fcntl.h>
#include <poll.h>
#include <unistd.h>

#include <array>
#include <algorithm>
#include <system_error>
#include <set>

#include "str.h"
#include "file.h"
#include "algo.h"
#include "exception.h"
#include "scope_guard.h"
#include "timeit.h"
#include "log.h"


extern char **environ;

namespace evm {

constexpr int kExecFailure = 127;
constexpr int kChildFailure = 126;

ProcessReturnCode::ProcessReturnCode(ProcessReturnCode &&p) noexcept
        : rawStatus_(p.rawStatus_) {
    p.rawStatus_ = ProcessReturnCode::RV_NOT_STARTED;
}

ProcessReturnCode &ProcessReturnCode::operator=(ProcessReturnCode &&p) noexcept {
    rawStatus_ = p.rawStatus_;
    p.rawStatus_ = ProcessReturnCode::RV_NOT_STARTED;
    return *this;
}

ProcessReturnCode::State ProcessReturnCode::state() const {
    if (rawStatus_ == RV_NOT_STARTED) return NOT_STARTED;
    if (rawStatus_ == RV_RUNNING) return RUNNING;
    if (WIFEXITED(rawStatus_)) return EXITED;
    if (WIFSIGNALED(rawStatus_)) return KILLED;
    throw std::runtime_error("Invalid ProcessReturnCode: " + std::to_string(rawStatus_));
}

void ProcessReturnCode::enforce(State expected) const {
    State s = state();
    if (s != expected) {
        throw std::logic_error("Bad use of ProcessReturnCode; state is "
                               + std::to_string(s) + " expected " + std::to_string(expected));
    }
}

int ProcessReturnCode::exitStatus() const {
    enforce(EXITED);
    return WEXITSTATUS(rawStatus_);
}

int ProcessReturnCode::killSignal() const {
    enforce(KILLED);
    return WTERMSIG(rawStatus_);
}

bool ProcessReturnCode::coreDumped() const {
    enforce(KILLED);
    return WCOREDUMP(rawStatus_);
}

string ProcessReturnCode::str() const {
    switch (state()) {
        case NOT_STARTED:
            return "not started";
        case RUNNING:
            return "running";
        case EXITED:
            return "exited with status " + std::to_string(exitStatus());
        case KILLED:
            return "killed by signal " + std::to_string(killSignal())
                   + (coreDumped() ? " (core dumped)" : "");
        default:  // unreached
            assert(false);
            return "";
    }
}

CalledProcessError::CalledProcessError(ProcessReturnCode rc)
        : returnCode_(rc), what_(returnCode_.str()) {
}

SubprocessSpawnError::SubprocessSpawnError(const char *executable, int errCode, int errnoValue)
        : errnoValue_(errnoValue),
          what_(string(errCode == kExecFailure ? "failed to execute " : "error preparing to execute ")
                + executable + ": " + str::strerror(errnoValue)) {
}

namespace {

// Copy pointers to the given strings in a format suitable for posix_spawn
unique_ptr<const char *[]> cloneStrings(const vector<string> &s) {
    std::unique_ptr<const char *[]> d(new const char *[s.size() + 1]);
    for (size_t i = 0; i < s.size(); i++) {
        d[i] = s[i].c_str();
    }
    d[s.size()] = nullptr;
    return d;
}

// Check a wait() status, throw on non-successful
void checkStatus(ProcessReturnCode returnCode) {
    if (returnCode.state() != ProcessReturnCode::EXITED ||
        returnCode.exitStatus() != 0) {
        throw CalledProcessError(returnCode);
    }
}

}  // namespace

Subprocess::Options &Subprocess::Options::fd(int fd, int action) {
    if (action == Subprocess::PIPE) {
        if (fd == 0) {
            action = Subprocess::PIPE_IN;
        } else if (fd == 1 || fd == 2) {
            action = Subprocess::PIPE_OUT;
        } else {
            throw invalid_argument("Only fds 0, 1, 2 are valid for action=PIPE: " + std::to_string(fd));
        }
    }
    fdActions_[fd] = action;
    return *this;
}

Subprocess::Subprocess(
        const vector<string> &argv,
        const Options &options,
        const char *executable,
        const vector<string> *env)
        : pid_(-1),
          returnCode_(RV_NOT_STARTED) {
    if (argv.empty())
        throw invalid_argument("argv must not be empty");
    if (!executable)
        executable = argv[0].c_str();

    spawn(cloneStrings(argv), executable, options, env);
}

Subprocess::Subprocess(
        const string &cmd,
        const Options &options,
        const vector<string> *env)
        : pid_(-1),
          returnCode_(RV_NOT_STARTED) {
    if (options.usePath_) {
        throw invalid_argument("usePath() not allowed when running in shell");
    }
    const char *shell = getenv("SHELL");
    if (!shell)
        shell = "/bin/sh";

    std::unique_ptr<const char *[]> argv(new const char *[4]);
    argv[0] = shell;
    argv[1] = "-c";
    argv[2] = cmd.c_str();
    argv[3] = nullptr;
    spawn(std::move(argv), shell, options, env);
}

Subprocess::~Subprocess() {
    // Subprocess cannot be destroyed without reaping child
    assert(returnCode_.state() != ProcessReturnCode::RUNNING);
}

namespace {

struct ChildErrorInfo {
    int errCode;
    int errnoValue;
};

void childError(int errFd, int errCode, int errnoValue) {
    ChildErrorInfo info = {errCode, errnoValue};
    // Write the error information over the pipe to our parent process.
    // We can't really do anything else if this write call fails.
    writeNoInt(errFd, &info, sizeof(info));
    // exit
    _exit(errCode);
}

}  // namespace

void Subprocess::setAllNonBlocking() {
    for (auto &p : pipes_) {
        int fd = p.pipe.fd();
        int flags = ::fcntl(fd, F_GETFL);
        if (flags == -1)
            throw system_error(errno, system_category(), "fcntl");
        int r = ::fcntl(fd, F_SETFL, flags | O_NONBLOCK);
        if (r == -1)
            throw system_error(errno, system_category(), "fcntl");
    }
}

void Subprocess::spawn(
        std::unique_ptr<const char *[]> argv,
        const char *executable,
        const Options &optionsIn,
        const vector<string> *env) {
    if (optionsIn.usePath_ && env) {
        throw invalid_argument("usePath() not allowed when overriding environment");
    }

    // Make a copy, we'll mutate options
    Options options(optionsIn);

    // On error, close all pipes_ (ignoring errors, but that seems fine here).
    auto pipesGuard = make_guard([this] { pipes_.clear(); });

    // Create a pipe to use to receive error information from the child,
    // in case it fails before calling exec()
    // Ask the child to close the read end of the error pipe.
    // Set the close-on-exec flag on the write side of the pipe.
    // This way the pipe will be closed automatically in the child if execve()
    // succeeds.  If the exec fails the child can write error information to the
    // pipe.
    int errFds[2];  // errFds[0] refers to the read end, errFds[1] refers to the write end
    if (::pipe2(errFds, O_CLOEXEC) == -1)
        throw system_error(errno, system_category(), "pipe2");

    auto close_err_fds = make_guard([&] {
        ::close(errFds[0]);
        if (errFds[1] >= 0)
            ::close(errFds[1]);
    });

    // Perform the actual work of setting up pipes then forking and
    // executing the child.
    spawnInternal(std::move(argv), executable, options, env, errFds[1]);

    // After spawnInternal() returns the child is alive.  We have to be very
    // careful about throwing after this point.  We are inside the constructor,
    // so if we throw the Subprocess object will have never existed, and the
    // destructor will never be called.
    //
    // We should only throw if we got an error via the errFd, and we know the
    // child has exited and can be immediately waited for.  In all other cases,
    // we have no way of cleaning up the child.

    // Close writable side of the errFd pipe in the parent process
    ::close(errFds[1]);
    errFds[1] = -1;

    // Read from the errFd pipe, to tell if the child ran into any errors before
    // calling exec()
    readChildErrorPipe(errFds[0], executable);

    // We have fully succeeded now, so release the guard on pipes_
    pipesGuard.dismiss();
}

void Subprocess::spawnInternal(
        std::unique_ptr<const char *[]> argv,
        const char *executable,
        Options &options,
        const vector<string> *env,
        int errFd) {
    // Parent work, pre-fork: create pipes
    vector<int> childFds;
    // Close all of the childFds as we leave this scope
    auto close_child_fds = make_guard([&] {
        // These are only pipes, closing them shouldn't fail
        for (int cfd : childFds)
            ::close(cfd);
    });

    int r;
    for (auto &p : options.fdActions_) {
        if (p.second == PIPE_IN || p.second == PIPE_OUT) {
            int fds[2];
            // We're setting both ends of the pipe as close-on-exec. The child
            // doesn't need to reset the flag on its end, as we always dup2() the fd,
            // and dup2() fds don't share the close-on-exec flag.
            // If possible, set close-on-exec atomically. Otherwise, a concurrent
            // Subprocess invocation can fork() between "pipe" and "fnctl",
            // causing FDs to leak.
            if (::pipe2(fds, O_CLOEXEC) == -1)
                throw system_error(errno, system_category(), "pipe2");

            pipes_.emplace_back();
            Pipe &pipe = pipes_.back();
            pipe.direction = p.second;
            int cfd;
            if (p.second == PIPE_IN) {
                pipe.pipe = File(fds[1], true);  // Parent gets writing end
                cfd = fds[0];  // Child gets reading end
            } else {
                pipe.pipe = File(fds[0], true);  // Parent gets reading end
                cfd = fds[1];  // Child gets writing end
            }
            p.second = cfd;  // ensure it gets dup2()ed
            pipe.childFd = p.first;
            childFds.push_back(cfd);
        }
    }

    // This should already be sorted, as options.fdActions_ is
    assert(std::is_sorted(pipes_.begin(), pipes_.end()));

    // Note that the const casts below are legit, per
    // http://pubs.opengroup.org/onlinepubs/009695399/functions/exec.html

    char **argVec = const_cast<char **>(argv.get());

    // Set up environment
    std::unique_ptr<const char *[]> envHolder;
    char **envVec;
    if (env) {
        envHolder = cloneStrings(*env);
        envVec = const_cast<char **>(envHolder.get());
    } else {
        envVec = environ;
    }

    // Block all signals around vfork; see http://ewontfix.com/7/.
    //
    // As the child may run in the same address space as the parent until
    // the actual execve() system call, any (custom) signal handlers that
    // the parent has might alter parent's memory if invoked in the child,
    // with undefined results.  So we block all signals in the parent before
    // vfork(), which will cause them to be blocked in the child as well (we
    // rely on the fact that Linux, just like all sane implementations, only
    // clones the calling thread).  Then, in the child, we reset all signals
    // to their default dispositions (while still blocked), and unblock them
    // (so the exec()ed process inherits the parent's signal mask)
    //
    // The parent also unblocks all signals as soon as vfork() returns.
    sigset_t allBlocked;
    r = sigfillset(&allBlocked);
    check_unix_error(r, "sigfillset");
    sigset_t oldSignals;

    r = pthread_sigmask(SIG_SETMASK, &allBlocked, &oldSignals);
    check_posix_error(r, "pthread_sigmask");
    SCOPE_EXIT {
                   // Restore signal mask
                   r = pthread_sigmask(SIG_SETMASK, &oldSignals, nullptr);
                   check_posix_error(r, "pthread_sigmask");
               };

    // Call c_str() here, as it's not necessarily safe after fork.
    const char *childDir =
            options.childDir_.empty() ? nullptr : options.childDir_.c_str();
    pid_t pid = vfork();
    if (pid == 0) {  // In child process.
        int errnoValue = prepareChild(options, &oldSignals, childDir);
        if (errnoValue != 0) {
            childError(errFd, kChildFailure, errnoValue);
        }

        errnoValue = runChild(executable, argVec, envVec, options);
        // If we get here, exec() failed.
        childError(errFd, kExecFailure, errnoValue);
    }
    // In parent process.  Make sure vfork() succeeded.
    check_unix_error(pid, "vfork");

    // Child is alive.  We have to be very careful about throwing after this
    // point.  We are inside the constructor, so if we throw the Subprocess
    // object will have never existed, and the destructor will never be called.
    //
    // We should only throw if we got an error via the errFd, and we know the
    // child has exited and can be immediately waited for.  In all other cases,
    // we have no way of cleaning up the child.
    pid_ = pid;
    returnCode_ = ProcessReturnCode(RV_RUNNING);
}

int Subprocess::prepareChild(const Options &options,
                             const sigset_t *sigmask,
                             const char *childDir) const {
    // While all signals are blocked, we must reset their
    // dispositions to default.
    for (int sig = 1; sig < NSIG; ++sig) {
        ::signal(sig, SIG_DFL);
    }

    {
        // Unblock signals; restore signal mask.
        int r = pthread_sigmask(SIG_SETMASK, sigmask, nullptr);
        if (r != 0) {
            return r;  // pthread_sigmask() returns an errno value
        }
    }

    // Change the working directory, if one is given
    if (childDir) {
        if (::chdir(childDir) == -1) {
            return errno;
        }
    }

    // We don't have to explicitly close the parent's end of all pipes,
    // as they all have the FD_CLOEXEC flag set and will be closed at
    // exec time.

    // Close all fds that we're supposed to close.
    for (auto &p : options.fdActions_) {
        if (p.second == CLOSE) {
            if (::close(p.first) == -1) {
                return errno;
            }
        } else if (p.second != p.first) {
            if (::dup2(p.second, p.first) == -1) {
                return errno;
            }
        }
    }

    // If requested, close all other file descriptors.  Don't close
    // any fds in options.fdActions_, and don't touch stdin, stdout, stderr.
    // Ignore errors.
    if (options.closeOtherFds_) {
        for (int fd = getdtablesize() - 1; fd >= 3; --fd) {
            if (options.fdActions_.count(fd) == 0) {
                ::close(fd);
            }
        }
    }

    // Opt to receive signal on parent death, if requested
    if (options.parentDeathSignal_ != 0) {
        if (prctl(PR_SET_PDEATHSIG, options.parentDeathSignal_, 0, 0, 0) == -1) {
            return errno;
        }
    }

    if (options.processGroupLeader_) {
        if (setpgrp() == -1) {
            return errno;
        }
    }

    return 0;
}

int Subprocess::runChild(const char *executable,
                         char **argv, char **env,
                         const Options &options) const {
    // Now, finally, exec.
    if (options.usePath_) {
        ::execvp(executable, argv);
    } else {
        ::execve(executable, argv, env);
    }
    return errno;  // Only return here when exec failed.
}

void Subprocess::readChildErrorPipe(int pfd, const char *executable) {
    ChildErrorInfo info;
    auto rc = readNoInt(pfd, &info, sizeof(info));
    if (rc == 0) {
        // No data means the child executed successfully, and the pipe
        // was closed due to the close-on-exec flag being set.
        return;
    } else if (rc != sizeof(ChildErrorInfo)) {
        // An error occurred trying to read from the pipe, or we got a partial read.
        // Neither of these cases should really occur in practice.
        //
        // We can't get any error data from the child in this case, and we don't
        // know if it is successfully running or not.  All we can do is to return
        // normally, as if the child executed successfully.  If something bad
        // happened the caller should at least get a non-normal exit status from
        // the child.
        log_error("unexpected error trying to read from child error pipe rc={}, errno={}", rc, errno);
        return;
    }

    // We got error data from the child.  The child should exit immediately in
    // this case, so wait on it to clean up.
    wait();

    // Throw to signal the error
    throw SubprocessSpawnError(executable, info.errCode, info.errnoValue);
}

ProcessReturnCode Subprocess::poll() {
    returnCode_.enforce(ProcessReturnCode::RUNNING);
    assert(pid_ > 0);
    int status;
    pid_t found = ::waitpid(pid_, &status, WNOHANG);
    // The spec guarantees that EINTR does not occur with WNOHANG, so the only
    // two remaining errors are ECHILD (other code reaped the child?), or
    // EINVAL (cosmic rays?), both of which merit an abort:
    check_unix_error(found, "waitpid(" + std::to_string(pid_) + ", &status, WNOHANG)");
    if (found != 0) {
        // Though the child process had quit, this call does not close the pipes
        // since its descendants may still be using them.
        returnCode_ = ProcessReturnCode(status);
        pid_ = -1;
    }
    return returnCode_;
}

bool Subprocess::pollChecked() {
    if (poll().state() == ProcessReturnCode::RUNNING) {
        return false;
    }
    checkStatus(returnCode_);
    return true;
}

ProcessReturnCode Subprocess::wait() {
    returnCode_.enforce(ProcessReturnCode::RUNNING);
    assert(pid_ > 0);
    int status;
    pid_t found;
    do {
        found = ::waitpid(pid_, &status, 0);
    } while (found == -1 && errno == EINTR);
    // The only two remaining errors are ECHILD (other code reaped the
    // child?), or EINVAL (cosmic rays?), and both merit an abort:
    check_unix_error(found, "waitpid(" + std::to_string(pid_) + ", &status, WNOHANG)");
    // Though the child process had quit, this call does not close the pipes
    // since its descendants may still be using them.
    assert(found == pid_);
    returnCode_ = ProcessReturnCode(status);
    pid_ = -1;
    return returnCode_;
}

void Subprocess::waitChecked() {
    wait();
    checkStatus(returnCode_);
}

void Subprocess::sendSignal(int signal) {
    returnCode_.enforce(ProcessReturnCode::RUNNING);
    int r = ::kill(pid_, signal);
    check_unix_error(r, "kill");
}

pid_t Subprocess::pid() const {
    return pid_;
}

namespace {

/*std::pair<const uint8_t *, size_t> queueFront(const IOBufQueue &queue) {
    auto *p = queue.front();
    if (!p) return std::make_pair(nullptr, 0);
    return io::Cursor(p).peek();
}*/

// fd write
//bool handleWrite(int fd, IOBufQueue &queue) {
bool handleWrite(int fd, vector<char> &queue) {
    for (; ;) {
        //auto p = queueFront(queue);
        //if (p.second == 0) {
        if (queue.empty()) {
            return true;  // EOF
        }

        //ssize_t n = writeNoInt(fd, p.first, p.second);
        ssize_t n = writeNoInt(fd, queue.data(), queue.size());
        if (n == -1 && errno == EAGAIN) {
            return false;
        }
        check_unix_error(n, "write");
        //queue.trimStart(n);
        queue.erase(queue.begin(), queue.begin() + n);
    }
}

// fd read
//bool handleRead(int fd, IOBufQueue &queue) {
bool handleRead(int fd, vector<char> &queue) {
    for (; ;) {
        //auto p = queue.preallocate(100, 65000);
        auto old_size = queue.size();
        if (queue.capacity() - queue.size() < 100)
            queue.resize(queue.size() + 65000);
        else
            queue.resize(queue.capacity());
        //ssize_t n = readNoInt(fd, p.first, p.second);
        ssize_t n = readNoInt(fd, queue.data() + old_size, queue.size() - old_size);
        if (n == -1 && errno == EAGAIN) {
            return false;
        }
        check_unix_error(n, "read");
        if (n == 0) {
            return true;
        }
        //queue.postallocate(n);
        queue.resize(old_size + n);
    }
}

bool discardRead(int fd) {
    static const size_t bufSize = 65000;
    // Thread unsafe, but it doesn't matter.
    static std::unique_ptr<char[]> buf(new char[bufSize]);

    for (; ;) {
        ssize_t n = readNoInt(fd, buf.get(), bufSize);
        if (n == -1 && errno == EAGAIN) {
            return false;
        }
        check_unix_error(n, "read");
        if (n == 0) {
            return true;
        }
    }
}

}  // namespace

//std::pair<string, string> Subprocess::communicate(StringPiece input) {
pair<string, string> Subprocess::communicate(string input) {
    //IOBufQueue inputQueue;
    //inputQueue.wrapBuffer(input.data(), input.size());
    vector<char> inputQueue(input.cbegin(), input.cend());

    auto outQueues = communicateIOBuf(inputQueue);
    /*auto outBufs = std::make_pair(outQueues.first.move(),
                                  outQueues.second.move());
    std::pair<string, string> out;
    if (outBufs.first) {
        outBufs.first->coalesce();
        out.first.assign(reinterpret_cast<const char *>(outBufs.first->data()),
                         outBufs.first->length());
    }
    if (outBufs.second) {
        outBufs.second->coalesce();
        out.second.assign(reinterpret_cast<const char *>(outBufs.second->data()),
                          outBufs.second->length());
    }*/
    std::pair<string, string> out;
    out.first.assign(outQueues.first.cbegin(), outQueues.first.cend());
    out.second.assign(outQueues.second.cbegin(), outQueues.second.cend());
    return out;
}

//std::pair<IOBufQueue, IOBufQueue> Subprocess::communicateIOBuf(IOBufQueue input) {
pair<vector<char>, vector<char>> Subprocess::communicateIOBuf(vector<char> &input) {
    // If the user supplied a non-empty input buffer, make sure
    // that stdin is a pipe so we can write the data.
    if (!input.empty()) {
        // findByChildFd() will throw invalid_argument if no pipe for
        // STDIN_FILENO exists
        findByChildFd(STDIN_FILENO);
    }

    //std::pair<IOBufQueue, IOBufQueue> out;
    pair<vector<char>, vector<char>> out;

    auto readCallback = [&](int pfd, int cfd) -> bool {
        if (cfd == STDOUT_FILENO) {
            return handleRead(pfd, out.first);
        } else if (cfd == STDERR_FILENO) {
            return handleRead(pfd, out.second);
        } else {
            // Don't close the file descriptor, the child might not like SIGPIPE,
            // just read and throw the data away.
            return discardRead(pfd);
        }
    };

    auto writeCallback = [&](int pfd, int cfd) -> bool {
        if (cfd == STDIN_FILENO) {
            return handleWrite(pfd, input);
        } else {
            // If we don't want to write to this fd, just close it.
            return true;
        }
    };

    communicate(std::move(readCallback), std::move(writeCallback));

    return out;  // RVO
}

void Subprocess::communicate(FdCallback readCallback,
                             FdCallback writeCallback) {
    // This serves to prevent wait() followed by communicate(), but if you
    // legitimately need that, send a patch to delete this line.
    returnCode_.enforce(ProcessReturnCode::RUNNING);
    setAllNonBlocking();

    vector<pollfd> fds;
    fds.reserve(pipes_.size());
    vector<size_t> toClose;  // indexes into pipes_
    toClose.reserve(pipes_.size());

    while (!pipes_.empty()) {
        fds.clear();
        toClose.clear();

        for (auto &p : pipes_) {
            pollfd pfd;
            pfd.fd = p.pipe.fd();
            // Yes, backwards, PIPE_IN / PIPE_OUT are defined from the
            // child's point of view.
            if (!p.enabled) {
                // Still keeping fd in watched set so we get notified of POLLHUP /
                // POLLERR
                pfd.events = 0;
            } else if (p.direction == PIPE_IN) {
                pfd.events = POLLOUT;
            } else {
                pfd.events = POLLIN;
            }
            fds.push_back(pfd);
        }

        int r;
        do {
            r = ::poll(fds.data(), fds.size(), -1);
        } while (r == -1 && errno == EINTR);
        check_unix_error(r, "poll");

        for (size_t i = 0; i < pipes_.size(); ++i) {
            auto &p = pipes_[i];
            auto parentFd = p.pipe.fd();
            assert(fds[i].fd == parentFd);
            short events = fds[i].revents;

            bool closed = false;
            if (events & POLLOUT) {
                assert(!(events & POLLIN));
                if (writeCallback(parentFd, p.childFd)) {
                    toClose.push_back(i);
                    closed = true;
                }
            }

            // Call read callback on POLLHUP, to give it a chance to read (and act
            // on) end of file
            if (events & (POLLIN | POLLHUP)) {
                assert(!(events & POLLOUT));
                if (readCallback(parentFd, p.childFd)) {
                    toClose.push_back(i);
                    closed = true;
                }
            }

            if ((events & (POLLHUP | POLLERR)) && !closed) {
                toClose.push_back(i);
                closed = true;
            }
        }

        // Close the fds in reverse order so the indexes hold after erase()
        for (auto rit = toClose.rbegin(); rit != toClose.rend(); ++rit) {
            auto pos = pipes_.begin() + *rit;
            pos->pipe.close();  // Throws on error
            pipes_.erase(pos);
        }
    }
}

void Subprocess::enableNotifications(int childFd, bool enabled) {
    pipes_[findByChildFd(childFd)].enabled = enabled;
}

bool Subprocess::notificationsEnabled(int childFd) const {
    return pipes_[findByChildFd(childFd)].enabled;
}

size_t Subprocess::findByChildFd(int childFd) const {
    auto pos = std::lower_bound(
            pipes_.begin(), pipes_.end(), childFd,
            [](const Pipe &pipe, int fd) {
                return pipe.childFd < fd;
            }
    );

    if (pos == pipes_.end() || pos->childFd != childFd) {
        throw invalid_argument("child fd not found " + std::to_string(childFd));
    }
    return pos - pipes_.begin();
}

void Subprocess::closeParentFd(int childFd) {
    size_t idx = findByChildFd(childFd);
    pipes_[idx].pipe.close();  // May throw
    pipes_.erase(pipes_.begin() + idx);
}

vector<Subprocess::ChildPipe> Subprocess::takeOwnershipOfPipes() {
    vector<Subprocess::ChildPipe> pipes;
    for (auto &p : pipes_) {
        pipes.emplace_back(p.childFd, std::move(p.pipe));
    }
    pipes_.clear();
    return pipes;
}

namespace {

class Initializer {
public:
    Initializer() {
        // We like EPIPE, thanks.
        ::signal(SIGPIPE, SIG_IGN);
    }
};

Initializer initializer;

}

namespace {

string execute(const string &cmd, int attempts, bool log, bool check,
               std::initializer_list<int> check_exit_status) {
    while (attempts > 0) {
        --attempts;
        Timeit timeit;
        Subprocess proc(cmd, Subprocess::pipeStdout()
                .stderr(STDOUT_FILENO) // redirect stderr to stdout
                .closeOtherFds());
        auto ret = proc.communicate(); // no stdin
        assert(ret.second.empty());
        auto rc = proc.wait();
        if (log)
            log_info("cmd '{}', returned {} in {}s, output '{}'",
                     cmd, rc.exitStatus(), timeit.get(), ret.first);

        if (algo::in(rc.exitStatus(), check_exit_status))
            return ret.first;

        if (attempts > 0) {
            static RandomSleep sleep(0.1, 1);
            sleep();
        }
        else {
            if (check)
                throw CalledProcessError(rc);
        }
    }

    return "";
}

}

string execute(const string &cmd, int attempts, bool log, bool check) {
    return execute(cmd, attempts, log, check, {0});
}

string execute(const string &cmd, int attempts, bool log, std::initializer_list<int> check_exit_status) {
    return execute(cmd, attempts, log, true, check_exit_status);
}

}
